package sunyu.toolkit.activemq;

import org.apache.activemq.pool.PooledConnection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import sunyu.toolkit.core.ExceptionKit;

import javax.jms.*;
import java.util.Enumeration;

/**
 * 消息接收者
 *
 * @author 孙宇
 */
public class ActiveMQReceiverTemplate implements MessageListener {

    private static final Logger logger = LoggerFactory.getLogger(ActiveMQReceiverTemplate.class);

    private String name;
    private Session session;
    private MessageConsumer consumer;

    /**
     * 初始化消费者
     *
     * @param name       消费者名称
     * @param connection 消费者连接
     * @param type       消费者类型
     * @param subject    主题
     */
    public ActiveMQReceiverTemplate(String name,
                                    PooledConnection connection,
                                    ActiveMQDestination type,
                                    String subject) {
        try {
            this.name = name;
            // 事务性会话，自动确认消息
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            // 消息的目的地（Queue/Topic）
            if (type.equals(ActiveMQDestination.Topic)) {
                Topic destination = session.createTopic(subject);
                consumer = session.createConsumer(destination);
            } else {
                Queue destination = session.createQueue(subject);
                consumer = session.createConsumer(destination);
            }
            consumer.setMessageListener(this);
        } catch (Exception e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
    }

    /**
     * 获得消费者名称
     *
     * @return 消费者名称
     */
    public String getName() {
        return name;
    }

    /**
     * 消费者监听
     *
     * @param message 消息
     */
    @Override
    public void onMessage(Message message) {
        try {
            if (message instanceof TextMessage) {
                TextMessage msg = (TextMessage) message;
                System.out.println(msg.getText());
            } else if (message instanceof MapMessage) {
                MapMessage msg = (MapMessage) message;
                Enumeration enumer = msg.getMapNames();
                while (enumer.hasMoreElements()) {
                    Object obj = enumer.nextElement();
                    System.out.println(msg.getObject(obj.toString()));
                }
            } else if (message instanceof StreamMessage) {
                StreamMessage msg = (StreamMessage) message;
                System.out.println(msg.readString());
                System.out.println(msg.readBoolean());
                System.out.println(msg.readLong());
            } else if (message instanceof ObjectMessage) {
                ObjectMessage msg = (ObjectMessage) message;
                System.out.println(msg);
            } else if (message instanceof BytesMessage) {
                BytesMessage msg = (BytesMessage) message;
                byte[] byteContent = new byte[1024];
                int length = -1;
                StringBuffer content = new StringBuffer();
                while ((length = msg.readBytes(byteContent)) != -1) {
                    content.append(new String(byteContent, 0, length));
                }
                System.out.println(content.toString());
            } else {
                System.out.println(message);
            }
        } catch (Exception e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
    }
}
